
import org.apache.spark.sql.Row
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types.{DataType, LongType, StringType, StructField, StructType}

/**
 * 自定义汇总函数
 * UDAF函数就是遍历表里的每一条数据，把每一条数据放入UDAF函数中做聚合
 */
class GroupConcatDistinct extends UserDefinedAggregateFunction {
  //UDAF输入类型为string
  override def inputSchema: StructType = {

    new StructType().add("cityInfo", StringType)

  }


  // 缓冲区的数据类型
  override def bufferSchema: StructType = {

    new StructType().add("bufferCityInfo", StringType)

  }

  // 输出数据类型
  override def dataType: DataType = StringType

  override def deterministic: Boolean = true

  // 初始化
  override def initialize(buffer: MutableAggregationBuffer): Unit = {
    buffer(0) = ""
  }

  override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    var bufferCityInfo = buffer.getString(0) // 缓冲区数据
    val cityInfo = input.getString(0) //输入数据

    if (!bufferCityInfo.contains(cityInfo)) {
      if ("".equals(bufferCityInfo)) {
        bufferCityInfo += cityInfo
      } else {
        bufferCityInfo += "," + cityInfo
      }
      buffer.update(0, bufferCityInfo) // 给buffer的0号位置，更新数据


    }
  }

  override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
    var bufferCityInfo1: String = buffer1(0).toString
    var bufferCityInfo2: String = buffer2(0).toString

    //两个buffer的数据相加
    for (cityInfo <- bufferCityInfo2.split(",")) {
      if(!bufferCityInfo1.contains(cityInfo)){
        if ("".equals(bufferCityInfo1)) {
          bufferCityInfo1 += cityInfo
        } else {
          bufferCityInfo1 += "," + cityInfo
        }
        buffer1.update(0,bufferCityInfo1 )
      }
    }
  }

  // 返回结果
  override def evaluate(buffer: Row): Any = {

    buffer.getString(0)
  }
}
